// Copyright (c) 2022 Presto Labs Pte. Ltd.
// Author: leon

#include "coin2/exchange/mexc_futures/feed_v1/processor.h"

#include <set>
#include <string_view>

#include "coin2/base/conversion.h"
#include "coin2/base/string_util.h"
#include "coin2/base/time.h"
#include "coin2/exchange/base/api_util/json_schema.h"
#include "coin2/exchange/mexc_futures/feed_v1/update.h"
#include "coin2/exchange/util/json_util.h"

namespace coin2::exchange::mexc_futures::feed_v1::impl {
void MexcFuturesFeedProcessor::OnTopicRecord(const TopicRecordPacket& packet) {
  upd_mgr_.onBeginPacket(packet);

  const auto& data = packet.payload;

  rapidjson::Document document;

  document.Parse(data.c_str());
  if (document.HasParseError()) {
    THROW() << "JSON parse error: "
            << GetParseError_En(document.GetParseError()) << packet.payload;
  }

  if (parser_config_.validate_schema() && schema_) {
    schema_->AcceptOrThrow(document);
  }

  const auto& channel = ExtractChannel(document);
  if (!CheckChannelFilter(channel)) {
    upd_mgr_.onEndPacket(packet);
    return;
  }
  if (document.HasMember("symbol")) {
    const std::string& native_symbol = document["symbol"].GetString();

    if (channel == "depth") {
      auto* book = upd_mgr_.GetBookMutable(native_symbol);
      if (book == nullptr) {
        upd_mgr_.onEndPacket(packet);
        return;
      }
      if (parser_config_.publish_bbo_book_only()) {
        std::optional<PriceQty> bid_pre = book->Bid0();
        std::optional<PriceQty> ask_pre = book->Ask0();
        MexcFuturesFeedUpdateParser::ParseIncremental(book, document);
        std::optional<PriceQty> bid_post = book->Bid0();
        std::optional<PriceQty> ask_post = book->Ask0();
        bool is_bbo_changed = (bid_pre != bid_post) || (ask_pre != ask_post);
        if (is_bbo_changed) {
          upd_mgr_.PublishBook();
        }
      } else {
        MexcFuturesFeedUpdateParser::ParseIncremental(book, document);
        upd_mgr_.PublishBook();
      }
    } else if (channel == "book_full") {
      auto* book = upd_mgr_.GetBookMutable(native_symbol);
      if (book == nullptr) {
        upd_mgr_.onEndPacket(packet);
        return;
      }
      if (parser_config_.publish_bbo_book_only()) {
        std::optional<PriceQty> bid_pre = book->Bid0();
        std::optional<PriceQty> ask_pre = book->Ask0();
        MexcFuturesFeedUpdateParser::ParseSnapshot(book, document);
        std::optional<PriceQty> bid_post = book->Bid0();
        std::optional<PriceQty> ask_post = book->Ask0();
        bool is_bbo_changed = (bid_pre != bid_post) || (ask_pre != ask_post);
        if (is_bbo_changed) {
          upd_mgr_.PublishBook();
        }
      } else {
        MexcFuturesFeedUpdateParser::ParseSnapshot(book, document);
        upd_mgr_.PublishBook();
      }
    } else if (channel == "trade") {
      if (!upd_mgr_.is_symbol_registered(native_symbol)) {
        upd_mgr_.onEndPacket(packet);
        return;
      }
      Trade* trade = upd_mgr_.GetTradeMutable();
      trade->Clear();
      MexcFuturesFeedUpdateParser::ParseTrade(document["data"], trade);
      upd_mgr_.PublishTrade(native_symbol, false);
    } else if (channel == "mark_price") {
      if (upd_mgr_.is_symbol_registered(native_symbol)) {
        auto* mark_price = upd_mgr_.GetMarkPriceMutable();
        mark_price->Clear();
        mark_price->timestamp = document["ts"].GetInt64() * 1'000'000LL;
        mark_price->mark_price = document["data"]["price"].GetDouble();
        upd_mgr_.PublishMarkPrice(native_symbol);
      }
    } else if (channel == "index") {
      if (upd_mgr_.is_symbol_registered(native_symbol)) {
        auto* index = upd_mgr_.GetIndexMutable();
        index->Clear();
        index->timestamp = document["ts"].GetInt64() * 1'000'000LL;
        index->price = document["data"]["price"].GetDouble();
        upd_mgr_.PublishIndex(native_symbol);
      }
    } else if (channel == "funding_rate") {
      if (upd_mgr_.is_symbol_registered(native_symbol)) {
        auto* funding_rate = upd_mgr_.GetFundingRateMutable();
        funding_rate->Clear();
        funding_rate->timestamp = document["ts"].GetInt64() * 1'000'000LL;
        funding_rate->funding_rate = document["data"]["rate"].GetDouble();
        funding_rate->funding_time = document["data"]["nextSettleTime"].GetInt64() * 1'000'000LL;
        upd_mgr_.PublishFundingRate(native_symbol);
      }
    } else {
      THROW() << "[Mexc Futures Feed Processor] unkown topic: " << packet.payload;
    }
  } else {
    THROW() << "[Mexc Futures Feed Processor] meta message: " << packet.payload;
  }
  upd_mgr_.onEndPacket(packet);
}
}  // namespace coin2::exchange::mexc_futures::feed_v1::impl
